package com.eshipenet.shipnet.equipmentextbuilder

import org.apache.flink.streaming.api.scala._
import play.api.libs.json.{JsObject, JsValue, Json}

object EquipmentExtStreamMergeTest {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val equipmentWithModelActionStream: DataStream[EquipmentWithModelAction] =
      createEquipmentWithModelActionStream(env)
    equipmentWithModelActionStream.print("equipmentWithModelActionStream")
    val s =
      Processor.mergeToEquipmentExtAction(equipmentWithModelActionStream)
    s.print()
    env.execute()
  }

  private def createEquipmentWithModelActionStream(
      env: StreamExecutionEnvironment): DataStream[EquipmentWithModelAction] = {
    env
      .socketTextStream("0.0.0.0", 4003, '\n', 10)
      .map(v => Json.parse(v))
      .map { v =>
        val op = (v \ "op").as[String]
        op match {
          case "i" =>
            val e     = Equipment((v \ "equipment").as[JsValue]).get
            val model = EquipmentModel((v \ "model").as[JsObject]).get
            val link  = EquipmentModelLink(e.id, model.id, Map(), Json.obj())
            val em    = EquipmentWithModel(e, model, link)
            Some(EquipmentWithModelInsert(em))
          case "d" =>
            val e     = Equipment((v \ "equipment").as[JsValue]).get
            val model = EquipmentModel((v \ "model").as[JsObject]).get
            val link  = EquipmentModelLink(e.id, model.id, Map(), Json.obj())
            val em    = EquipmentWithModel(e, model, link)
            Some(EquipmentWithModelDelete(em))
          case _ => None
        }
      }
      .filter(_.isDefined)
      .map(_.get)
  }

}
